package com.we.risk.phoneRecord.var;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.doris.SinkDoris;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.phoneRecord.common.PhoneRecordDataUtils;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import com.we.risk.phoneRecord.stage2.adm.AdmUserLabel;
import com.we.utils.ParamCheck;
import org.apache.commons.io.FileUtils;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;

import static com.we.risk.phoneRecord.var.dimphone.TagCallRecord.getNameTags;

public class AtoCallRecord {
    public static final String RELEASEPROP =
            "risk/phonerecord/var/adm_user_label_kfk_prod.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmUserLabel.class);
    public static final String SPLITWORD = "#";
    public static final String DATEFORMAT = "yyyy-MM-dd HH:mm:ss";
    public static final String INPUTKEYBY = "user_key";
    public static final String RECVTIME = "recv_time";
    public static final String OUTKEYBY = "user_key";
    public static final String LABELOUTIME = "label_output_time";
    public static final String OUTJOBID = "job_id";
    public static final String JOBID = "jobID";
    public static final String PHONECALLLIST = "phoneCallList";
    public static final String REMOVEJSONKEY = "numberType";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(5 * 1000); // ms
            ckConf.setCheckpointTimeout(5* 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            int tmpKfkPartitions = paramReader.getTmpKfkPartitions();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            /** distributed cache for Sougou+Cuishou Tag + mob_area_clean Data */

            env.registerCachedFile(
                    Stage2GlobalSetting.V1MOBAREAISPFORMAT, Stage2GlobalSetting.V1MOBAREAISPFORMATFILENAME);
            env.registerCachedFile(
                    Stage2GlobalSetting.SGCSTAGDATA, Stage2GlobalSetting.SGCSCATCHFILENAME);

            env.registerCachedFile(
                    Stage2GlobalSetting.CSTAGV2DATA, Stage2GlobalSetting.CSV2CATCHFILENAME);

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("PhoneRecord-var-source")
                            .setParallelism(tmpKfkPartitions);


            SingleOutputStreamOperator<String> out = input
                    .filter(
                    new FilterFunction<String>() {
                        @Override
                        public boolean filter(String value) throws Exception {
                            JSONObject inputJson = JSONObject.parseObject(value);
                            if (inputJson.containsKey(INPUTKEYBY)
                                    && inputJson.getString(INPUTKEYBY) != null) {
                                return true;
                            }
                            return false;
                        }
                    })
                    .keyBy(
                            new KeySelector<String, String>() {
                                @Override
                                public String getKey(String value) throws Exception {
                                    JSONObject inputJson = JSONObject.parseObject(value);
                                    return inputJson.getString(INPUTKEYBY);
                                }
                            })
                    .flatMap(new AtoCallRecordRichFlatMapFunc());
            DorisSink<String> dorisSink = SinkDoris.buidJsonArrayDorisSink("10.10.148.68:8030,10.10.148.69:8030,10.10.148.74:8030",
                    "ato_var_callrecord.var_230701",
                    "cr_var_230701",
                    "doris",
                    "wetc@#dw23",
                    "recv_time");

            out.sinkTo(dorisSink).setParallelism(2);

            env.execute(AtoCallRecord.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class AtoCallRecordRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> dimUserVState;
        ValueState<Integer> callListSizeVState;
        SimpleDateFormat sdf;
        HashMap<String, String> mobAreaIspMap = new HashMap<>();
        HashMap<String, String> sgcsTagMap = new HashMap<>();
        HashMap<String, String> csTagV2Map = new HashMap<>();
        HashMap<String, Integer> mobileOptMap = new HashMap<>();
        HashMap<String, Integer> callTypeMap = new HashMap<>();

        @Override
        public void open(Configuration parameters) throws Exception {
            dimUserVState =
                getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("dimUserVState", JSONObject.class));
            callListSizeVState =
                getRuntimeContext().getState(new ValueStateDescriptor<Integer>("callListSize", Integer.class));
            sdf = new SimpleDateFormat(DATEFORMAT);
            initMobileAreaIspMap();
            initSgCsTagMap();
            initCsTagV2Map();
            initMobileOptMap();
            initCallTypeMap();
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                JSONObject tmpUserState = dimUserVState.value();
                JSONObject curUserState = (tmpUserState == null ? new JSONObject() : tmpUserState);

                if (value.contains(JOBID)) {
                    //phoneCalllist
                    if (curUserState.size() > 1)
                    {
                        JSONArray res = ProcessData(value, curUserState, callListSizeVState);
                        out.collect(res.toString());
                    }
                } else {
                    //userInfo
                    LOG.debug("Debug: dim UserKey come in: " + value);
                    JSONObject inputJson = JSONObject.parseObject(value);
                    if (!curUserState.equals(inputJson)) {
                        dimUserVState.update(inputJson);
                    }
                }
            } catch (Exception e) {
                LOG.error("AtoCallRecordRichFlatMapFunc Exception: " + e.toString());
                LOG.error("AtoCallRecordRichFlatMapFunc excep: " + value);
                e.printStackTrace(System.out);
            }
        }

        private JSONArray ProcessData(String value, JSONObject userState, ValueState<Integer> callListSizeVState) throws Exception {
            JSONArray resJsonArray = new JSONArray();

            Integer tmpCallListSize = callListSizeVState.value();
            int lastCallListSize = (tmpCallListSize == null ? 0 : tmpCallListSize);

            JSONObject inputJson = JSONObject.parseObject(value);
            JSONObject commonJson = new JSONObject();
            commonJson.put(OUTKEYBY, inputJson.getString(INPUTKEYBY));
            commonJson.put(OUTJOBID, inputJson.getString(JOBID));
            commonJson.put(RECVTIME, inputJson.getLong(RECVTIME));

            JSONArray phoneCallList = inputJson.getJSONArray(PHONECALLLIST);
            ParamCheck.isNull(phoneCallList, PHONECALLLIST);
            int size = phoneCallList.size();

            for (int i = 0; i < size; i++) {
                JSONObject callDetail = phoneCallList.getJSONObject(i);
                callDetail.putAll(commonJson);
                callDetail.put("row_num", i);
                callDetail.put("is_active", 1);

                formatCallDetail(callDetail);

                formatSamePositions(callDetail, userState);
                resJsonArray.add(callDetail);
            }

            if(size < lastCallListSize) {
                for (int i = size; i < lastCallListSize; i++) {
                    JSONObject expiredCall = new JSONObject();
                    expiredCall.putAll(commonJson);
                    expiredCall.put("row_num", i);
                    expiredCall.put("is_active", 0);

                    resJsonArray.add(expiredCall);
                }
                callListSizeVState.update(size);
            }

            return resJsonArray;
        }

        private void formatCallDetail(JSONObject callJson) {
            if (callJson.containsKey(REMOVEJSONKEY)) {
                callJson.remove(REMOVEJSONKEY);
            }

            Integer call_type = callTypeMap.get(callJson.getString("type"));
            callJson.put("call_type",(call_type == null ? 0 : call_type));
            if (callJson.containsKey("type")) {
                callJson.remove("type");
            }

            Long call_time = callJson.getLong("date");
            callJson.put("call_time", sdf.format(new Date(call_time)));
            if(callJson.containsKey("date")) {
                callJson.remove("date");
            }

            String name = callJson.getString("name");
            String number = callJson.getString("number");

            String nameTags = getNameTags(name, number);
            callJson.put("name_tags", nameTags);

            String phone = PhoneRecordDataUtils.getVarFormatedPhone(number);
            String mobile = PhoneRecordDataUtils.getVarMobile(phone);
            String phone_tags = PhoneRecordDataUtils.getVarPhoneTags(number);
            String f_phone = PhoneRecordDataUtils.forMattedPhoneNumber(number);

            callJson.put("phone", phone);
            callJson.put("mobile", mobile);
            callJson.put("phone_tags", phone_tags);
            callJson.put("f_phone", f_phone);

            if (f_phone != null) {
                consSgCsTag(f_phone, sgcsTagMap, csTagV2Map, callJson);
                if (f_phone.length() == 11) {
                    String subNumber = f_phone.substring(0, 7);
                    consMobAreaIsp(subNumber, mobAreaIspMap, callJson);
                }
            }


            if (callJson.containsKey("number")) {
                callJson.remove("number");
            }
        }
        private void formatSamePositions(JSONObject callDetail, JSONObject userState) {
            byte samePidProv = 0;
            byte samePidCity = 0;
            byte sameMobProv = 0;
            byte sameMobCity = 0;
            byte sameAddProv = 0;
            byte sameAddCity = 0;
            byte sameMobOpt = 0;

            if (callDetail.containsKey("mobile_prov")) {
                String mob_prov = callDetail.getString("mobile_prov");
                String state_mob_prov = userState.getString("mob_prov");
                String state_pid_prov = userState.getString("pid_prov");
                String state_ad_prov = userState.getString("ad_prov");

                sameMobProv = (byte) (mob_prov.equals(state_mob_prov) ? 1 : 0);
                samePidProv = (byte) (mob_prov.equals(state_pid_prov) ? 1 : 0);
                sameAddProv = (byte) (mob_prov.equals(state_ad_prov) ? 1 : 0);
            }

            if (callDetail.containsKey("mobile_city")) {
                String mob_city = callDetail.getString("mobile_city");
                String state_mob_city = userState.getString("mob_city");
                String state_pid_city = userState.getString("pid_city");
                String state_ad_city = userState.getString("ad_city");

                sameMobCity = (byte) (mob_city.equals(state_mob_city) ? 1 : 0);
                samePidCity = (byte) (mob_city.equals(state_pid_city) ? 1 : 0);
                sameAddCity = (byte) (mob_city.equals(state_ad_city) ? 1 : 0);
            }

            if (callDetail.containsKey("mobile_opt")) {
                String mob_opt = callDetail.getString("mobile_opt");
                String state_mob_isp = userState.getString("mob_isp");
                sameMobOpt = (byte) (mob_opt.equals(state_mob_isp) ? 1: 0);
            }

            callDetail.put("same_positions", ""+samePidProv+samePidCity+sameMobProv+sameMobCity+sameMobOpt+sameAddProv+sameAddCity);
        }

        //清洗过后的prov+city
        private void initMobileAreaIspMap() throws IOException {
            File mobAreaIsp = getRuntimeContext()
                    .getDistributedCache()
                    .getFile(Stage2GlobalSetting.V1MOBAREAISPFORMATFILENAME);
            List<String> mobAreaLines = FileUtils.readLines(mobAreaIsp);
            for (String line : mobAreaLines) {
                String[] splits = line.split(",");
                if (splits.length == 4) {
                    String mobile = splits[0];
                    String province = splits[1];
                    String city = splits[2];
                    String isp = splits[3];
                    /** 维表异常数据处理 */
                    if (province == null || province.equals("")) {
                        province = "null";
                    }

                    if (city == null || city.equals("")) {
                        city = "null";
                    }

                    if (isp == null || isp.equals("")) {
                        isp = "null";
                    }

                    String mapKey = mobile;
                    String mapVal = province + SPLITWORD + city + SPLITWORD + isp;
                    mobAreaIspMap.put(mapKey, mapVal);
                } else {
                    LOG.error("DimPhoneRecordLabel initV1CleanMap error read line: " + line);
                }
            }
        }

        private void initMobileOptMap() {
            mobileOptMap.put("中国联通", 1);
            mobileOptMap.put("中国移动", 2);
            mobileOptMap.put("中国电信", 3);
        }

        private void initCallTypeMap() {
//            callTypeMap.put("UNKNOWN", 0);
            callTypeMap.put("INCOMING", 1);
            callTypeMap.put("MISSED", 2);
            callTypeMap.put("OUTGOING", 3);
            callTypeMap.put("REJECTED", 4);
            callTypeMap.put("BLOCKED", 5);
        }
        private void initCsTagV2Map() throws IOException {
            File sgTagFile =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.CSV2CATCHFILENAME);
            List<String> sgTagLines = FileUtils.readLines(sgTagFile);
            for (String line : sgTagLines) {
                csTagV2Map.put(line,"");
            }
        }

        private void initSgCsTagMap() throws IOException {
            File sgTagFile =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.SGCSCATCHFILENAME);
            List<String> sgTagLines = FileUtils.readLines(sgTagFile);
            for (String line : sgTagLines) {
                String[] splits = line.split(",");
                if (splits.length == 3) {
                    String phone_num = splits[0];
                    String sgTag = splits[1];
                    String coName = splits[2];
                    String mapKey = phone_num;
                    String mapVal = sgTag + SPLITWORD + coName;
                    sgcsTagMap.put(mapKey, mapVal);
                } else {
                    LOG.error("DimPhoneRecordLabel initSgCsTagMap error read line: " + line);
                }
            }
        }

        private void consSgCsTag(String mobile,
                                 HashMap<String, String> sgcsMap,
                                 HashMap<String, String> csTagV2Map,
                                 JSONObject outJson) {
            try {
                String sgcsTag = sgcsMap.get(mobile);
                if (sgcsTag != null) {
                    String[] splits = sgcsTag.split(SPLITWORD);
                    if (splits.length == 2) {
                        outJson.put("phone_sgtag", splits[0]);
                        outJson.put("phone_cstag", splits[1]);
                    }
                }
                outJson.put("phone_cstag_v2", (csTagV2Map.containsKey(mobile) ? 1 : 0));
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("consSgCsTag exception input mobile: " + mobile);
            }
        }

        /** 返回映射完全的地址+ISP运营商 */
        private void consMobAreaIsp(String mobile, HashMap<String, String> areaMap, JSONObject outJson) {
            try {
                String res = areaMap.get(mobile);
                if (res != null) {
                    LOG.debug("MapMobArea get res: " + res);
                    String[] splits = res.split(SPLITWORD);
                    if (splits.length == 3) {
                        outJson.put("mobile_prov", splits[0]);
                        outJson.put("mobile_city", splits[1]);
                        Integer isp_code = mobileOptMap.get(splits[2]);
                        outJson.put("mobile_opt", (isp_code == null ? 0 : isp_code));
                    }
                }
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("consMobAreaIsp exception input mobile: " + mobile);
            }
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTKEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }


}
